package cn.iocoder.mall.admin.mq;

import cn.iocoder.common.utils.common.api.vo.SMSMQDTO;
import cn.iocoder.common.utils.common.util.json.JacksonUtils;
import cn.iocoder.mall.admin.client.SmsClient;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;

/***
 *@author yinxucun
 *@date 2019-11-27 18:18 
 */

public class SmsConsumer {

    @Value("${mall.rocket.ip}")
    String  rocketIp;

    @Resource(name = "smsAliYunClient")
    SmsClient client;
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("mall_user_verification_code");

    public SmsConsumer() throws MQClientException {

        consumer.setNamesrvAddr(rocketIp);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    for (MessageExt messageExt : msgs) {
                        String message = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                        SMSMQDTO smsmqdto = (SMSMQDTO) JacksonUtils.parseObject(message, SMSMQDTO.class);
                        client.singleSendVerificationCode(smsmqdto.getCode(), smsmqdto.getMobile());
                    }
                }catch (Exception e){
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}
